package com.snailhd.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;

@Slf4j
@Configuration
public class MqttConfiguration implements CommandLineRunner {

    public final static String  PRE_LOG_PUBLISH = "发送消息";
    public final static String  PRE_LOG_SUBSCRIBE = "订阅消息";
    public final static String  PRE_LOG_UNSUBSCRIBE = "取消订阅";

    @Value("${snailhd.mqtt.serverURI}")
    private String serverURI;
    @Value("${snailhd.mqtt.clientId}")
    private String clientId;
    @Value("${snailhd.mqtt.topic}")
    private String topic;

    private MqttClient mqttClient;
    private static MemoryPersistence memoryPersistence;
    private static MqttConnectOptions mqttConnectOptions;




    @Override
    public void run(String... args) throws Exception {
        connect();
        subscribe(topic);

    }

    @Scheduled(cron = "${snailhd.mqtt.scheduled.cron}")
    public void Scheduled(){
        connect();
    }

    //	发布消息
    public void publish(String topic,String message,int qos) {
        exec(PRE_LOG_PUBLISH, new MqttExcutor(){
            @Override
            public void exec(MqttClient client) throws Exception {
                MqttMessage mqttMessage = new MqttMessage();
                mqttMessage.setQos(qos);
                mqttMessage.setPayload(message.getBytes());
                MqttDeliveryToken publish = mqttClient.getTopic(topic).publish(mqttMessage);
                if(!publish.isComplete()) {
                    log.info("发送消息成功:{}->{}",topic,message);
                }else{
                    log.error("发送消息失败:{}->{}",topic,message);
                }
            }
        });
    }

    //订阅
    public void subscribe(String topic) {
        exec(PRE_LOG_SUBSCRIBE, new MqttExcutor(){
            @Override
            public void exec(MqttClient client) throws Exception {
                mqttClient.subscribe(topic, 1);
            }
        });
    }


    //取消订阅
    public void cleanTopic(String topic) {
        exec(PRE_LOG_UNSUBSCRIBE, new MqttExcutor(){
            @Override
            public void exec(MqttClient client) throws Exception {
                mqttClient.unsubscribe(topic);
            }
        });
    }


    public void exec(String prelog,MqttExcutor excutor){
        for (int i=0;i<5&&!mqttClient.isConnected();i++){
            if(i>0){
                try { Thread.sleep(1000); } catch (InterruptedException e) { }
            }
            connect();
        }
        if(mqttClient.isConnected()){
            try {
                excutor.exec(mqttClient);
            }catch (Exception e){
                log.error("{}执行异常:",prelog,e);
            }
        }
    }




    //	重新连接
    public void connect() {
        if(mqttClient==null){
            try {
                memoryPersistence = new MemoryPersistence();
                mqttClient = new MqttClient(serverURI, clientId, memoryPersistence);
                mqttClient.setCallback(new MqttRecieveCallback());
                connect();
            } catch (MqttException e) {
                log.error("serverURI:{},clientId:{} 连接异常",serverURI,clientId,e);
            }
        }
        boolean connected = mqttClient.isConnected();
        log.info("mqttClient 连接检查:{}",connected);
        if(!connected) {
            try {
                mqttClient.connect(mqttConnectOptions);
                log.info("mqttClient 连接成功!serverURI:{},clientId:{}",serverURI,clientId);
            } catch (MqttException e) {
                log.error("mqttClient 连接异常!serverURI:{},clientId:{}",serverURI,clientId,e);
            }
        }
    }

    //	关闭连接
    public void closeConnect() {
        //关闭存储方式
        if(null != memoryPersistence) {
            try {
                memoryPersistence.close();
            } catch (MqttPersistenceException e) {
                log.error("关闭 memoryPersistence 异常！",e);
            }
        }
        //关闭连接
        if(null != mqttClient && mqttClient.isConnected()) {
            try {
                mqttClient.disconnect();
                mqttClient.close();
            } catch (MqttException e) {
                log.error("关闭 mqttClient 异常！",e);
            }
        }
    }

    static {
        mqttConnectOptions = new MqttConnectOptions();
        //true可以安全地使用内存持久性作为客户端断开连接时清除的所有状态
        mqttConnectOptions.setCleanSession(true);
        //设置连接超时
        mqttConnectOptions.setConnectionTimeout(30);
    }

    interface MqttExcutor{
        void exec(MqttClient client) throws Exception;
    }
}
